/*
 *  Copyright (C) 2010 Zubanov Dmitry
 * 
 *  This program is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 * 
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 * 
 *  You should have received a copy of the GNU General Public License
 *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.jssdb.core.net;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import org.jssdb.core.event.CustomEvent;
import org.jssdb.core.event.CustomEventListener;
import org.jssdb.core.josql.OQLHandler;
import org.jssdb.core.josql.OQLParser;
import org.jssdb.core.query.Query;
import org.jssdb.core.ram.RamHandler;
import org.jssdb.core.utils.JSQLExeption;
import org.jssdb.core.utils.SizeException;

/**
 *
 * @author developer
 */
public class Server {

    private String host = "localhost";
    private int port = 5500;
    private DataChunkCollector dataChunkCollector = DataChunkCollector.getInstance();
    private static final Logger logger = Logger.getLogger(Server.class.getName());
    private RamHandler ramHandler = RamHandler.getInstance();
    OQLHandler DBQLlHandler = OQLHandler.getInstance();
    OQLParser DBQLParser = OQLParser.getInstance();

    private Server() {
        dataChunkCollector.addCustomEventListener(new CustomEventListener() {

            @Override
            public void customEventOccurred(CustomEvent evt) {
                if (evt.getSource() instanceof CarrierRequestResponse) {
                    carrierRequestResponseHandler((CarrierRequestResponse) evt.getSource());
                }
            }
        });
    }

    private void carrierRequestResponseHandler(CarrierRequestResponse crr) {
        if (crr.getOQLRequest() != null) {
            try {
                ArrayList<Map> result = DBQLlHandler.getResultSet(DBQLParser.getParameters(crr.getOQLRequest()), true);
                crr.setResultObject(result);
            } catch (JSQLExeption ex) {
                logger.log(Level.SEVERE, crr.getOQLRequest(), ex);
            }

        } else if(crr.getRequest() != null){
            try {
                ArrayList<Map<Long, Object>> result = Query.getInstance().getResultSet(crr.getRequest());
                crr.setResultObject(result);
            } catch (SizeException ex) {
                Logger.getLogger(Server.class.getName()).log(Level.SEVERE, null, ex);
            }

        } else {
            if (crr.getAction() == CarrierRequestResponse.ADD) {
                long id = -1;
                try {
                    id = RamHandler.getInstance().addObject(crr.getTransferObject());
                } catch (Exception e) {
                    logger.log(Level.WARNING, null, e);
                }

                if (id != -1) {
                    crr.setStatus(CarrierRequestResponse.ADDED_OK);
                    crr.setResultObject(id);
                } else {
                    crr.setStatus(CarrierRequestResponse.ADDED_FALSE);
                    crr.setResultObject(id);
                }
            }

            if (crr.getAction() == CarrierRequestResponse.MODIFY) {
                if (crr.getTransferObjectId() != -1) {
                    RamHandler.getInstance().updateObject(crr.getTransferObjectId(), crr.getTransferObject());
                    crr.setStatus(CarrierRequestResponse.MODIFIED_OK);
                }
            }

            if (crr.getAction() == CarrierRequestResponse.DELETE) {
                if (crr.getTransferObjectId() != -1) {
                    boolean result = RamHandler.getInstance().deleteObject(crr.getTransferObjectId(),
                            "" + crr.getTransferObject());
                    if (result) {
                        crr.setStatus(CarrierRequestResponse.DELETED_OK);
                    } else {
                        crr.setStatus(CarrierRequestResponse.DELETED_FALSE);
                    }
                }
            }
        }

        senderThread(crr.getMyHost(), crr.getMyPort(), crr);
    }

    public void setPort(int port) {
        this.port = port;
    }

    public void setBindHost(String host) {
        this.host = host;
    }

    public void start() {
        Executors.newSingleThreadExecutor().execute(reciever());
    }

    public Runnable reciever() {
        Runnable r = new Runnable() {

            @Override
            public void run() {
                try {
                    DatagramChannel channel = DatagramChannel.open();
                    DatagramSocket socket = channel.socket();
                    InetAddress addr = InetAddress.getByName(host);
                    SocketAddress address = new InetSocketAddress(addr, port);
                    socket.bind(address);

                    ByteBuffer buffer = ByteBuffer.allocateDirect(65536);
                    while (true) {
                        SocketAddress client = channel.receive(buffer);
                        buffer.flip();

                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes, 0, bytes.length);
                        ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                        GZIPInputStream gz = new GZIPInputStream(bais);
                        ObjectInputStream in = new ObjectInputStream(gz);
                        Object object = in.readObject();
                        logger.log(Level.INFO, "Server reciever get object ({0})", object);

                        if (object instanceof DataChunk) {
                            dataChunkCollector.addChunk((DataChunk) object);
                        }

                        buffer.clear();
                    }
                } catch (Exception e) {
                    logger.log(Level.SEVERE, host, e);
                }
            }
        };

        return r;
    }

    public void senderThread(String remoteHost, int remotePort, Object object) {
        Executors.newSingleThreadExecutor().execute(sender(remoteHost, remotePort, object));
    }

    private Runnable sender(final String remoteHost, final int remotePort, final Object object) {
        Runnable r = new Runnable() {

            @Override
            public void run() {
                try {
                    DatagramChannel channel = DatagramChannel.open();
                    channel.configureBlocking(false);
                    channel.connect(new InetSocketAddress(remoteHost, remotePort));
                    Selector selector = Selector.open();
                    channel.register(selector, SelectionKey.OP_WRITE);

                    while (true) {
                        selector.select(1000);
                        Set readyKeys = selector.selectedKeys();
                        if (readyKeys.isEmpty()) {
                            break;

                        } else {
                            Iterator iterator = readyKeys.iterator();
                            while (iterator.hasNext()) {
                                SelectionKey key = (SelectionKey) iterator.next();
                                iterator.remove();

                                if (key.isWritable()) {
                                    ArrayList<DataChunk> dataChunkList = new DataChunkList().getDataChunkList(object);
                                    for (DataChunk dataChunk : dataChunkList) {
                                        ByteArrayOutputStream baos = new ByteArrayOutputStream();
                                        GZIPOutputStream gz = new GZIPOutputStream(baos);
                                        ObjectOutputStream oout = new ObjectOutputStream(gz);
                                        oout.writeObject(dataChunk);
                                        oout.flush();
                                        oout.close();
                                        gz.flush();
                                        gz.close();

                                        channel.write(ByteBuffer.wrap(baos.toByteArray()));
                                    }
                                }
                                key.interestOps(SelectionKey.OP_READ);
                            }
                        }
                    }

                    selector.close();
                    channel.close();
                } catch (Exception e) {
                    logger.log(Level.SEVERE, null, e);
                }
            }
        };

        return r;
    }

    public static Server getInstance() {
        return ServerHolder.INSTANCE;
    }

    private static class ServerHolder {

        private static final Server INSTANCE = new Server();
    }
}
